package com.qupeng.demo.kafka.kafkaapache.consumer;

import com.qupeng.demo.kafka.kafkaapache.producer.Product;
import org.apache.kafka.common.serialization.Deserializer;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;

public class CustomizedDeserializer implements Deserializer<Product> {
    @Override
    public Product deserialize(String topic, byte[] data) {
        ByteBuffer byteBuffer = ByteBuffer.wrap(data);
        int id = byteBuffer.getInt();
        int nameSize = byteBuffer.getInt();
        byte[] nameBytes = new byte[nameSize];
        byteBuffer.get(nameBytes);
        String name = new String(nameBytes, StandardCharsets.UTF_8);
        return new Product(id, name);
    }
}
